package com.wy.sloth.core;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;

import com.alibaba.fastjson.JSONObject;
import com.wy.sloth.config.SlothConfig;
import com.wy.sloth.constant.SlothConstants;
import com.wy.sloth.exception.SlothException;
import com.wy.sloth.model.SlothMessage;
import com.wy.sloth.model.SlothReceipt;
import com.wy.sloth.model.SlothReceiptData;
import com.wy.sloth.redis.service.SlothRedisService;
import com.wy.sloth.util.StringUtils;

import lombok.Getter;
import lombok.Setter;

/**
 * 消息接收器，一个应用只可以建一个消息接收器
 * 
 * @author wangyuan
 * @since 2019年1月31日
 */
public class SlothStandardReceiver implements SlothReceiver {
    
    private static final Logger logger = LoggerFactory.getLogger(SlothStandardReceiver.class);
    
    private static final Map<String, SlothMsgHandler> HANDLERS = new HashMap<>();
    
    /**
     * 消息处理器线程池数量，默认10
     */
    @Getter
    @Setter
    private int corePoolSize = 10;
    
    /**
     * 此应用对应的消息队列名称
     */
    private String msgQueueName;

    /**
     * 接收器线程池
     */
    private ExecutorService receiverExecutor;
    
    /**
     * 处理器线程池
     */
    private ExecutorService handlerExecutor;
    
    @Getter
    @Setter
    private ListOperations<String, JSONObject> listOperations;
    
    @Getter
    @Setter
    private ZSetOperations<String, String> zSetOperations;
    
    @Getter
    @Setter
    private HashOperations<String, String, JSONObject> hashOperations;
    
    /**
     * 项目相关配置
     */
    @Getter
    @Setter
    private SlothConfig slothConfig;
    
    /**
     * 项目关于redis的封装操作服务类
     */
    @Getter
    private SlothRedisService slothRedisService;
    
    /**
     * 统计当前消息接收器接收到的并且正在处理的消息数
     */
    private final AtomicInteger activeMsgCount = new AtomicInteger(0);
    
    /**
     * 当前项目名称生成器
     */
    private SlothNameManager slothNameManager;
    
    /**
     * 是否自动注册type与消息队列的映射关系到redis
     */
    @Setter
    @Getter
    private boolean autoRegisterTypeAndQueueRelation = true;
    
    private SlothMsgLogManager slothMsgLogManager;
    
    private RedisTemplate<String, JSONObject> redisTemplate;
    
    private RedisTemplate<String, String> stringRedisTemplate;
    
    public SlothStandardReceiver(RedisTemplate<String, JSONObject> redisTemplate,
            RedisTemplate<String, String> stringRedisTemplate, 
            SlothConfig slothConfig) {
    	this.redisTemplate = redisTemplate;
    	this.stringRedisTemplate = stringRedisTemplate;
        this.listOperations = redisTemplate.opsForList();
        this.hashOperations = redisTemplate.opsForHash();
        this.zSetOperations = stringRedisTemplate.opsForZSet();
        this.slothConfig = slothConfig;
    }
    
    @Override
    public void start() {
        logger.info("-----消息接收者准备启动-----");
        //创建实例
        this.slothNameManager = new SlothStandardNameManager(slothConfig);
        this.slothRedisService = new SlothRedisService(redisTemplate, stringRedisTemplate, slothConfig);
        this.slothMsgLogManager = new SlothMsgLogManager(stringRedisTemplate, slothRedisService, slothConfig, slothNameManager);
        
        //创建线程池实例
        receiverExecutor = Executors.newSingleThreadExecutor(new ReceiverExecutorThreadFactory());
        handlerExecutor = Executors.newFixedThreadPool(corePoolSize, new HandlerExecutorThreadFactory());
        
        //生成消息队列名称
        msgQueueName = slothNameManager.getApplicationQueueKey();
        logger.info("本应用创建的消息队列：{}", msgQueueName);
        
        //运行接收者工作线程
        receiverExecutor.execute(new Receiver());
        
        //先清理本消息队列与type的所有注册关系，然后再重新注册
        logger.info("清理队列{}与所有type的映射关系", msgQueueName);
        slothRedisService.clearTypeAndQueueMapping(msgQueueName);
        //注册type与本消息队列的关系到redis
        if(autoRegisterTypeAndQueueRelation){
            registerTypeAndQueueRelation();
        }
        
        //启动日志管理器
        slothMsgLogManager.start();
        logger.info("-----消息接收者结束启动-----");
    }
    
    /**
     * 注册type与本消息队列的关系到redis
     */
    public void registerTypeAndQueueRelation() {
        if(slothRedisService == null){
            throw new SlothException("Instance Of SlothRedisService Not Be Null");
        }
        if(msgQueueName == null){
            throw new SlothException("MsgQueueName Not Be Null");
        }
        
        for(String typeItem : HANDLERS.keySet()){
            logger.info("注册消息类型为{}与队列{}的映射关系至redis", typeItem, msgQueueName);
            slothRedisService.registerTypeAndQueueNameRelation(typeItem, msgQueueName);
        }
    }
    
    @Override
    public void destroy() {
        logger.info("-----消息接收者准备终止-----");
        //注销type与消息队列映射关系
        for(String typeItem : HANDLERS.keySet()){
            slothRedisService.deleteTypeAndQueueNameRelation(typeItem, msgQueueName);
        }
        
        //停掉线程池
        if(!receiverExecutor.isShutdown()){
            receiverExecutor.shutdown();
        }
        if(!handlerExecutor.isShutdown()){
            handlerExecutor.shutdown();
        }
        
        //停止日志管理器的工作
        slothMsgLogManager.destroy();
        logger.info("-----消息接收者结束终止-----");
    }
    
    @Override
    protected void finalize() throws Throwable {
        if(!receiverExecutor.isShutdown()){
            receiverExecutor.shutdown();
        }
        if(!handlerExecutor.isShutdown()){
            handlerExecutor.shutdown();
        }
        super.finalize();
    }
    
    /**
     * 注册消息处理器
     * @param type
     * @param handler
     */
    public void registerHandler(String type, SlothMsgHandler handler) {
        HANDLERS.put(type, handler);
    }

    /**
     * 消息接收
     * 
     * @author wangyuan
     * @since 2019年2月1日
     */
    class Receiver implements Runnable {

        @Override
        public void run() {
            try {
                while(!Thread.currentThread().isInterrupted()){
                    //如果未注册队列与type之间的映射关系，则不进行逻辑处理
                    if(!autoRegisterTypeAndQueueRelation){
                        continue;
                    }
                    
                    receive();
                }
            } catch (Exception e) {
                logger.error("接收并处理消息时出现错误，本次失败原因是：", e);
            }
        }
        
        /**
         * 接收消息并将相应的消息分配给相应的处理器去执行
         */
        private void receive() {
            JSONObject msgObj = null;
            String msgId = null;
            try {
                //如果处理器线程池现在属于满载状态，则不去获取消息
                if(activeMsgCount.get() >= corePoolSize){
                    TimeUnit.MICROSECONDS.sleep(500L);
                    return;
                }
                
                //去消息队列中获取消息
                msgObj = listOperations.rightPop(msgQueueName, 800L, TimeUnit.MILLISECONDS);
                if(msgObj == null){
                    return;
                }
                
                //msgId为空，则不处理
                msgId = msgObj.getString("id");
                if(StringUtils.isEmpty(msgId)){
                    return;
                }
                
                SlothMessage message = JSONObject.toJavaObject(msgObj, SlothMessage.class);
                if(StringUtils.isEmpty(message.getType())){
                    logger.error("type不能为空");
                    slothRedisService.updateDealStatusOfMsgForLog(msgId, msgQueueName, SlothConstants.STATUS_EXEC_FAIL, "type不能为空");
                    return;
                }
                
                SlothMsgHandler handler = HANDLERS.get(message.getType());
                if(handler == null){
                    logger.error("找不到对应的消息处理器，type为{}", message.getType());
                    slothRedisService.updateDealStatusOfMsgForLog(msgId, msgQueueName, SlothConstants.STATUS_EXEC_FAIL, "找不到对应的消息处理器");
                    return;
                }
                
                logger.info("接收到的消息id为{}，type为{}，timeStamp为{}", message.getId(), message.getType(), message.getTimeStamp());
                handlerExecutor.execute(new Handler(handler, message));
            } catch (InterruptedException e) {
                logger.warn("处理消息时抛出中断异常{}", e.getClass().getName());
            } catch (Exception e) {
                if(msgObj == null){
                    logger.warn("处理消息（message = null）时出现错误，原因是：{}", e.getClass().getName());
                } else {
                    logger.error("处理消息（" + msgObj.toJSONString() + "）时出现错误，原因是：", e);
                    slothRedisService.updateDealStatusOfMsgForLog(msgId, msgQueueName, SlothConstants.STATUS_EXEC_FAIL, "系统错误");
                }
            }
        }
    }
    
    /**
     * 消息处理线程
     * 
     * @author wangyuan
     * @since 2019年2月1日
     */
    class Handler implements Runnable {
        
        /**
         * 消息对应的消息处理者
         */
        private final SlothMsgHandler handler;
        /**
         * 需要被处理的消息
         */
        private final SlothMessage message;
        
        public Handler(SlothMsgHandler handler, SlothMessage message) {
            this.handler = handler;
            this.message = message;
            
            //创建任务时，就增加在处理的任务统计量
            activeMsgCount.incrementAndGet();
        }

        @Override
        public void run() {
            logger.info("type为{}的消息处理器开始执行", message.getType());
            try {
            	beforeHandle();
            	JSONObject handleResult = handler.handle(message.getData());
            	afterHandleWhenSuccess(handleResult);
            } catch (Exception e) {
                logger.error("处理type为" + message.getType() + "的消息时出现错误，原因是：", e);
                afterHandleWhenCatchException(e);
            } finally {
                //任务处理完成之后，减少统计量
                activeMsgCount.decrementAndGet();
            }
            logger.info("type为{}的消息处理器结束执行", message.getType());
        }
        
        private void beforeHandle() {
        	slothRedisService.updateDealStatusOfMsgForLog(message.getId(), msgQueueName, SlothConstants.STATUS_EXECUTING);
        }
        
        private void afterHandleWhenSuccess(JSONObject handleResult) {
        	slothRedisService.updateDealStatusOfMsgForLog(message.getId(), msgQueueName, SlothConstants.STATUS_EXEC_SUCCESS);
            if(message.isSync()){
            	SlothReceiptData receiptData = new SlothReceiptData(message.getId(), msgQueueName);
                receiptData.setStatus(SlothReceipt.STATUS_SUCCESS);
                receiptData.setReceiptData(handleResult);
                
                hashOperations.put(slothNameManager.getMsgReceiptKey(), receiptData.getHashKey(), receiptData.toJsonObject());
            }
        }
        
        private void afterHandleWhenCatchException(Exception e) {
        	slothRedisService.updateDealStatusOfMsgForLog(message.getId(), msgQueueName, SlothConstants.STATUS_EXEC_FAIL, e.getMessage());
            if(message.isSync()){
            	SlothReceiptData receiptData = new SlothReceiptData(message.getId(), msgQueueName);
                receiptData.setStatus(SlothReceipt.STATUS_ERROR);
                receiptData.setErrorMsg(e.getMessage());
                
                hashOperations.put(slothNameManager.getMsgReceiptKey(), receiptData.getHashKey(), receiptData.toJsonObject());
            }
        }
        
    }
    
    static class HandlerExecutorThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(0);
        private final String namePrefix;
        {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = "slothHandlerPool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
              if (t.isDaemon()) {
                  t.setDaemon(false);
              }
              
              if (t.getPriority() != Thread.NORM_PRIORITY) {
                  t.setPriority(Thread.NORM_PRIORITY);
              }
              
              return t;
        }
    }
    
    static class ReceiverExecutorThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(0);
        private final String namePrefix;
        {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = "slothReceiverPool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
              if (t.isDaemon()) {
                  t.setDaemon(false);
              }
              
              if (t.getPriority() != Thread.NORM_PRIORITY) {
                  t.setPriority(Thread.NORM_PRIORITY);
              }
                  
              return t;
        }
        
    }
}
